package com.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.app.function.CustomerDeserialization;
import com.app.function.DimSinkFunction;
import com.app.function.TableProcessFunction;
import com.bean.TableProcess;
import com.common.GlobalConfig;
import com.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2022/1/3 9:49
 */

//TODO: 数据流 web/app -> Nginx -> Springboot  -> Mysql  ->  FlinkApp  -> Kafka(ods) -> FlinkApp -> Kafka/Phoenix
//TODO: 程序 mockDb -> FlinkCDC -> Kafka(ZK) -> BaseDBApp -> Kafka/Phoenix(HBase,ZK,HDFS)
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //TODO: 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //TODO:Flink-CDC 将读取 binlog 的位置信息以状态的方式保存在 CK,如果想要做到断点续传,需要从 Checkpoint 或者 Savepoint 启动程序
            /*//开启CK并指定状态后端为FS menory fs rocksdb
            env.setStateBackend(new FsStateBackend("hdfs://192.168.88.109:9820/gmall-flink/ck"));
            //开启 Checkpoint,每隔 5 秒钟做一次 CK
            env.enableCheckpointing(5000L);
            //指定 CK 的一致性语义
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(10000L);
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
            //设置任务关闭的时候保留最后一次 CK 数据
            env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
            //指定从 CK 自动重启策略
            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,2000L));
            //设置访问 HDFS 的用户名
            System.setProperty("HADOOP_USER_NAME", "root");*/

        //TODO: 2.消费kafka ods_base_db 主题数据创建流
        String sourceTopic = "ods_base_db";
        String groupId = "base_db_app" + GlobalConfig.NUMBER;
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));

        //TODO: 3.将每行数据转换为JSON对象并过滤（delete）  主流
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject)
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String type = jsonObject.getString("type");
                        /*if (type.equals("delete")){
                            return false;
                        }
                        return true;*/
                        return !type.equals("delete");
                    }
                });

        //TODO: 4.使用 FlinkCDC 消费配置表并处理成  广播流
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("192.168.88.109")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("gmall-realtime")
                .tableList("gmall-realtime.table_process")
                .startupOptions(StartupOptions.initial())
                .deserializer(new CustomerDeserialization())
                .build();

        DataStreamSource<String> tableProcessStrDS = env.addSource(sourceFunction);
        MapStateDescriptor mapStateDescriptor = new MapStateDescriptor<String, TableProcess>
                ("map-state", String.class, TableProcess.class);
        BroadcastStream<String> broadcastStream = tableProcessStrDS.broadcast(mapStateDescriptor);

        //TODO: 5.连接 主流 和 广播流
        BroadcastConnectedStream<JSONObject, String> connectedStream = jsonObjDS.connect(broadcastStream);

        //TODO: 6.分流  处理数据 广播流数据 主流数据（根据广播流进行处理）
        OutputTag<JSONObject> hbaseTag = new OutputTag<JSONObject>("hbase-tag") {
        };
        SingleOutputStreamOperator<JSONObject> kafka = connectedStream.process(new TableProcessFunction(hbaseTag, mapStateDescriptor));

        //TODO: 7.提取Kafka流数据 和 Hbase流数据
        DataStream<JSONObject> hbase = kafka.getSideOutput(hbaseTag);


        //TODO: 8.将 Kafka数据写入到Kafka主题，将 Hbase数据写入到Phoenix表
        System.out.println("任务开始>>>>>>>>>>>>>>>>>>");
        kafka.print("Kafka>>>>>>>>>>>");
        hbase.print("HBase>>>>>>>>>>>");
        hbase.addSink(new DimSinkFunction());
        /*//不可用
        kafka.map(jsonObj ->jsonObj.toJSONString())
                .addSink(MyKafkaUtil.getKafkaProducer(""));*/
        kafka.addSink(MyKafkaUtil.getKafkaProducer(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject element, @Nullable Long timestamp) {
                return new ProducerRecord<byte[], byte[]>(element.getString("sinkTable"),
                        element.getString("after").getBytes());
            }
        }));


        //TODO: 9.启动任务
        env.execute("BaseDBApp");
    }
}
